package org.wyz.mapreduce.joinwithcache;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

public class JoinWithCacheMapper extends Mapper<LongWritable, Text, Text, NullWritable> {

    private static final String cacheFileName = "user_exam.csv";
    private static HashMap<String, String> cache = new HashMap<>();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        // 如果是空的，就把缓存读出来
        // 这里可以看到，每个map任务都要把文件处理一遍，这是冗余操作
        // 优化思路：把文件处理完再发到缓存
        if (cache.isEmpty()) {
            URI cacheFileURI = context.getCacheFiles()[0];
            String cacheFilePath = cacheFileURI.getPath();
            BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(cacheFilePath)));
            String line;
            while ((line = reader.readLine()) != null) {
                List<String> idAndAttrs = seperate(line, 0, ",");
                cache.put(idAndAttrs.get(0), idAndAttrs.get(1));
            }
        }


        // 然后就可以开始了
        String currentFileName = ((FileSplit) context.getInputSplit()).getPath().getName();
        if (!cacheFileName.equals(currentFileName)) {
            List<String> idAndAttrs = seperate(value.toString(), 0, ",");
            String id = idAndAttrs.get(0);
            String attrs = idAndAttrs.get(1);
            if (cache.containsKey(id)) {
                context.write(new Text(id + "\t" + cache.get(id) + "," + attrs), NullWritable.get());
            }
        }

    }


    /**
     * 从行中抽取出id 和 剩下的属性
     *
     * @param row
     * @return 第一个是id，第二个是剩下的attr
     */
    public List<String> seperate(String row, int idPosition, String seperator) {
        String[] attrs = row.split(seperator);
        String id = attrs[idPosition];
        List<String> res = new ArrayList<>();
        res.add(id);
        // 把剩下的拼起来
        StringBuilder sb = new StringBuilder("");
        for (int i = 0; i < attrs.length; i++) {
            if (i == idPosition) {
                continue;
            }
            sb.append(attrs[i]);
            if (i < attrs.length - 1) {
                sb.append(seperator);
            }
        }
        res.add(sb.toString());
        return res;
    }
}
